package com.cmcc.shared.utils;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

public class RxUtil {

    private static final int DEFAULT_DURATION_TIME = 2;

    public static <T> void subscribe(Observable<T> observable, Observer<T> observer) {
        //noinspection unchecked
        doForAndroid(observable)
                .subscribe(observer);
    }

    public static <T> Disposable subscribe(Observable<T> observable, Consumer<T> onNext) {
        //noinspection unchecked
        return doForAndroid(observable)
                .subscribe(onNext);
    }

    public static <T> Disposable subscribe(Observable<T> observable, Consumer<T> onNext, Consumer<T> onError) {
        //noinspection unchecked
        return doForAndroid(observable)
                .subscribe(onNext, onError);
    }

    public static <T> Disposable subscribe(Observable<T> observable, Consumer<T> onNext, Consumer<T> onError, Action onComplete) {
        //noinspection unchecked
        return doForAndroid(observable)
                .subscribe(onNext, onError, onComplete);
    }

    //指定时间内只会发出一次请求
    public static <T> void throttle(Observable<T> observable, Observer<T> observer) {
        throttle(observable, observer, DEFAULT_DURATION_TIME);
    }

    //指定时间内只会发出一次请求
    public static <T> Observable<T> throttle(Observable<T> observable) {
        //noinspection unchecked
        return doForAndroid(observable)
                .throttleFirst(DEFAULT_DURATION_TIME, TimeUnit.SECONDS);
    }

    //指定时间内只发出一次请求
    public static <T> void throttle(Observable<T> observable, Observer<T> observer, int windowDuration) {
        //noinspection unchecked
        doForAndroid(observable)
                .throttleFirst(windowDuration, TimeUnit.SECONDS)
                .subscribe(observer);
    }

    public static Disposable postDelay(int delay, TimeUnit timeUnit, Consumer<Long> consumer) {
        return Observable.timer(delay, timeUnit).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()).subscribe(consumer);
    }

    private static <T> Observable doForAndroid(Observable<T> observable) {
        return observable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

}
